use std::{
    fs,
    os::fd::{AsFd, AsRawFd},
    path::Path,
    sync::{Arc, LazyLock},
};

use crate::{
    error::*,
    iface::{AppHandleIface, AppManager, EmptyAppHandle},
    runtime::RuntimeManager,
    stream::Stream,
    utils::socket_ext::set_ipv6only,
};
use arc_swap::{ArcSwap, ArcSwapOption};
use conf::ListenerConf;
use log::{debug, error};
use parking_lot::Mutex;
use tokio::{
    net::{TcpListener, TcpSocket, UnixListener},
    select,
};
use tokio_util::sync::CancellationToken;

pub mod conf;
mod test;

pub enum RawListener {
    TcpListener(TcpListener),
    UnixListener(UnixListener),
}

#[allow(dead_code)]
impl RawListener {
    async fn accept(&self) -> Result<Stream> {
        match self {
            Self::TcpListener(t) => {
                let res = t.accept().await.if_err(
                    ErrorType::AcceptError,
                    "tcp listener accept error",
                )?;
                Ok(res.0.into())
            }

            Self::UnixListener(u) => {
                let res = u.accept().await.if_err(
                    ErrorType::AcceptError,
                    "unix listener accept error",
                )?;
                Ok(res.0.into())
            }
        }
    }
}

pub(crate) type ArcListener = Arc<Listener>;

pub struct Listener {
    conf: Arc<ListenerConf>,
    raw_listener: ArcSwapOption<RawListener>,
    cancel_token: CancellationToken,
    handle: ArcSwap<AppHandleIface>,
}

#[allow(dead_code)]
impl Listener {
    pub fn new_tcp(conf: Arc<ListenerConf>) -> Result<Self> {
        let listener_type = &conf.listener_type;
        let listener_address = &conf.address;
        let app_name = conf.app_name.clone();

        if listener_type != "tcp" {
            return Err(Error::from_msg(
                ErrorType::InvalidError,
                format!(
                    "Expect tcp listener but got {}, address: {}",
                    listener_type, &conf.address
                ),
            ));
        }

        let addr_parse: Result<std::net::SocketAddr> =
            listener_address.parse().if_err(
                ErrorType::InvalidError,
                format!("invalid listen address, address: {}", &conf.address),
            );
        let addr = match addr_parse {
            Ok(addr) => addr,
            Err(e) => {
                return Err(e);
            }
        };

        let mut is_ipv4 = false;
        let socket = if addr.is_ipv4() {
            is_ipv4 = true;
            TcpSocket::new_v4()
        } else {
            TcpSocket::new_v6()
        }
        .if_err(
            ErrorType::CreateError,
            format!("create socket error, address: {}", &conf.address),
        )?;

        if conf.nodelay {
            socket.set_nodelay(true).if_err(
                ErrorType::SetSockoptError,
                format!("set nodelay failed, address: {}", &conf.address),
            )?;
        }

        if conf.reuseaddr {
            socket.set_reuseaddr(true).if_err(
                ErrorType::SetSockoptError,
                format!("set reuseaddr failed, address: {}", &conf.address),
            )?;
        }

        if conf.reuseport {
            socket.set_reuseport(true).if_err(
                ErrorType::SetSockoptError,
                format!("set reuseport failed, address: {}", &conf.address),
            )?;
        }

        let sock_fd = socket.as_fd().as_raw_fd();

        if conf.ipv6only && !is_ipv4 {
            set_ipv6only(sock_fd);
        }

        socket.bind(addr).if_err(
            ErrorType::BindError,
            format!("socket bind error, address: {}", &conf.address),
        )?;
        debug!("bind {} success", &conf.address);

        let mut backlog = conf.backlog as u32;
        if backlog < 128 {
            backlog = 128;
        }

        let ls = socket.listen(backlog).if_err(
            ErrorType::ListenError,
            format!("socket listen error, address: {}", &conf.address),
        )?;
        debug!("listen {} success", &conf.address);

        Ok(Listener {
            conf,
            raw_listener: ArcSwapOption::new(Some(Arc::new(
                RawListener::TcpListener(ls),
            ))),
            cancel_token: CancellationToken::new(),
            handle: ArcSwap::new(AppManager::get_app_handle(&app_name)),
        })
    }

    pub fn new_unix(conf: Arc<ListenerConf>) -> Result<Self> {
        let empty_handle = Box::new(EmptyAppHandle {});
        let path = Path::new(&conf.address);
        if path.exists() {
            _ = fs::remove_file(path);
        }

        let u = UnixListener::bind(path).if_err(
            ErrorType::BindError,
            format!("unix listener bind failed, address: {}", &conf.address),
        )?;
        debug!("bind {} success", &conf.address);

        Ok(Listener {
            conf,
            raw_listener: ArcSwapOption::new(Some(Arc::new(
                RawListener::UnixListener(u),
            ))),
            cancel_token: CancellationToken::new(),
            handle: ArcSwap::new(Arc::new(empty_handle)),
        })
    }

    #[inline]
    pub fn get_conf(&self) -> &ListenerConf {
        &self.conf
    }

    #[inline]
    pub fn get_raw_listener(&self) -> Option<Arc<RawListener>> {
        self.raw_listener.load().clone()
    }

    pub async fn run(self: Arc<Self>) {
        let l = self.raw_listener.load().clone().unwrap();
        let worker_runtime = RuntimeManager::get_worker_runtime();
        let cancel_token = self.cancel_token.clone();
        loop {
            let res = select! {
                res = l.accept() => res,
                _ = cancel_token.cancelled() => break,
            };

            let stream = match res {
                Ok(s) => s,
                Err(e) => {
                    error!("accept error: {:?} in {:?}", e, &self.conf.address);
                    return;
                }
            };

            let handle = self.handle.load().clone();
            let self_clone = self.clone();
            worker_runtime.get_handle().spawn(async move {
                handle.handle(stream, self_clone).await;
            });
        }
    }
}

pub struct ListenerManager {
    listeners: Mutex<Vec<ArcListener>>,
}

#[allow(dead_code)]
impl ListenerManager {
    pub fn get_instance() -> &'static ListenerManager {
        static LISTENER_MANAGER: LazyLock<ListenerManager> =
            LazyLock::new(|| ListenerManager {
                listeners: Mutex::new(Vec::new()),
            });
        &LISTENER_MANAGER
    }

    pub fn get_listener_by_address(address: &str) -> Option<ArcListener> {
        let instance = Self::get_instance();
        let listeners = instance.listeners.lock();
        for v in listeners.iter() {
            if v.conf.address == address {
                return Some(v.clone());
            }
        }

        None
    }

    pub fn get_listeners_by_app_name(app_name: &str) -> Vec<ArcListener> {
        let mut vector = Vec::new();
        let instance = Self::get_instance();
        let listeners = instance.listeners.lock();
        for v in listeners.iter() {
            if v.conf.app_name == app_name {
                vector.push(v.clone());
            }
        }

        vector
    }

    pub fn register_listener_by_conf(conf: Arc<ListenerConf>) -> Result<()> {
        let instance = Self::get_instance();

        let listener_type = &conf.listener_type;

        let l;
        if listener_type == "tcp" {
            l = Listener::new_tcp(conf)?;
        } else if listener_type == "unix" {
            l = Listener::new_unix(conf)?;
        } else {
            unimplemented!("Invalid listener type");
        }

        instance.listeners.lock().push(Arc::new(l));

        Ok(())
    }

    pub fn run() {
        let instance = ListenerManager::get_instance();
        let worker_runtime = RuntimeManager::get_worker_runtime();
        let guard_listeners = instance.listeners.lock();

        for v in guard_listeners.iter() {
            let v_clone = v.clone();
            worker_runtime.get_handle().spawn(async move {
                v_clone.run().await;
            });
        }
    }

    pub fn stop() {
        let instance = ListenerManager::get_instance();
        let guard_listeners = instance.listeners.lock();

        for v in guard_listeners.iter() {
            v.cancel_token.cancel();
        }
    }
}
